package com.lazar.module1.subject5

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}

import scala.collection.mutable

/**
  * 跨节点(网络)与worker通信
  * 接收注册
  * 保存注册信息
  * 回复worker注册成功！！
  */
class SparkMaster extends Actor {
  val workers = mutable.HashMap[String, WorkerInfo]()

  override def receive: Receive = {
    //启动
    case "start" => {
      println("master 已经启动。。。")
      self ! StartTimeOutWorker
    }
    //接收注册信息
    case RegisterWorkerInfo(id, cpu, ram) => {
      //是否已经注册过了
      if (!workers.contains(id)) {
        //保存worker信息
        val workerInfo = new WorkerInfo(id, cpu, ram)
        workers += (id -> workerInfo)
        //回复客户端注册成功
        sender() ! RegisteredWorkerInfo
        println(s"注册成功，workerid=$id")
      }
    }
    case HeartBeat(id) => {
      //更新 id 对应的 worker 的心跳
      if (workers.contains(id)) {
        workers(id).lastHeartBeatTime = System.currentTimeMillis()
        println(s"workerid=$id 更新心跳成功~") }
    }

    case StartTimeOutWorker =>{
      //启动定时器
      val actorSystem: ActorSystem = context.system
      import actorSystem.dispatcher
      import scala.concurrent.duration._
      context.system.scheduler.schedule(0 millis, 10000 millis, self, RemoveTimeOutWorker)
    }

    case RemoveTimeOutWorker => {
      //定时清理超时 6s 的 worker,scala
      //获取当前的时间
      val currentTime = System.currentTimeMillis()
      val workersInfo = workers.values //获取到所有注册的 worker 信息
      //先将超时的一次性过滤出来，然后对过滤到的集合一次性删除
      workersInfo.filter( currentTime - _.lastHeartBeatTime > 6000 ).foreach(
        workerInfo=>{ workers.remove(workerInfo.id)
        })
      printf("当前有%d 个 worker 存活\n", workers.size)
    }
  }
}

object SparkMaster extends App {
  //跨节点(网络)与worker通信
  val masterHost = "127.0.0.1"
  val masterPort = 10000

  val config: Config = ConfigFactory.parseString(
    s"""
akka.actor.provider="akka.remote.RemoteActorRefProvider"
akka.remote.netty.tcp.hostname=$masterHost
akka.remote.netty.tcp.port=$masterPort
""".stripMargin
  )

  //创建actorsystem
  val sparkMasterActorSystem = ActorSystem("SparkMaster", config)
  //注册actor
  private val sparkMaster0Ref: ActorRef = sparkMasterActorSystem.actorOf(Props[SparkMaster],
    "SparkMaster01")
  //启动masteractor
  sparkMaster0Ref ! "start"
}
